package com.electric.flinkcep.serialization;

import com.electric.flinkcep.model.Event;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * 事件反序列化器
 * 用于将Kafka中的JSON数据反序列化为Event对象
 */
public class EventDeserializer implements DeserializationSchema<Event> {
    
    private static final Logger logger = LoggerFactory.getLogger(EventDeserializer.class);
    private final ObjectMapper objectMapper;
    
    public EventDeserializer() {
        this.objectMapper = new ObjectMapper();
    }
    
    @Override
    public Event deserialize(byte[] message) throws IOException {
        if (message == null || message.length == 0) {
            logger.warn("接收到空消息，跳过反序列化");
            return null;
        }
        
        try {
            String jsonString = new String(message, StandardCharsets.UTF_8);
            logger.debug("反序列化消息: {}", jsonString);
            
            Event event = objectMapper.readValue(jsonString, Event.class);
            
            // 验证必要字段
            if (event.getId() == null || event.getType() == null) {
                logger.warn("事件缺少必要字段 (id或type), 跳过处理: {}", jsonString);
                return null;
            }
            
            // 设置默认时间戳（如果没有提供）
            if (event.getTimestamp() == 0) {
                event.setTimestamp(System.currentTimeMillis());
            }
            
            // 设置默认严重级别（如果没有提供）
            if (event.getSeverity() == null || event.getSeverity().trim().isEmpty()) {
                event.setSeverity("INFO");
            }
            
            logger.debug("成功反序列化事件: {}", event);
            return event;
            
        } catch (IOException e) {
            logger.error("反序列化消息失败: {}", new String(message, StandardCharsets.UTF_8), e);
            throw e;
        } catch (Exception e) {
            logger.error("处理消息时发生未知错误: {}", new String(message, StandardCharsets.UTF_8), e);
            throw new IOException("反序列化失败", e);
        }
    }
    
    @Override
    public boolean isEndOfStream(Event nextElement) {
        // 流不会结束
        return false;
    }
    
    @Override
    public TypeInformation<Event> getProducedType() {
        return TypeInformation.of(Event.class);
    }
}